block/internal/da/interface.go (1)

20-23: Clarify the post-cancellation drain contract in the comment.

“Callers MUST drain the channel after cancellation” can be interpreted differently; consider explicitly stating the expected sequence (cancel context, keep receiving until channel closes).

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@block/internal/da/interface.go` around lines 20 - 23, Update the comment for
Subscribe to explicitly describe the post-cancellation drain sequence: state
that callers should cancel the provided ctx, then continue receiving from the
returned <-chan datypes.SubscriptionEvent (e.g., range over the channel) until
it is closed by the implementation; clarify that the channel will be closed once
cancellation is observed and that draining (receiving until close) prevents
goroutine/resource leaks. Reference the Subscribe(ctx context.Context, namespace
[]byte) signature and datypes.SubscriptionEvent in the comment.

block/internal/syncing/da_retriever_tracing.go (1)

67-69: Consider instrumenting ProcessBlobs in the tracing wrapper.

This new hot path currently emits no span/attributes, which limits visibility into inline blob processing latency and output volume.

Suggested refactor

 func (t *tracedDARetriever) ProcessBlobs(ctx context.Context, blobs [][]byte, daHeight uint64) []common.DAHeightEvent {
-	return t.inner.ProcessBlobs(ctx, blobs, daHeight)
+	ctx, span := t.tracer.Start(ctx, "DARetriever.ProcessBlobs",
+		trace.WithAttributes(
+			attribute.Int64("da.height", int64(daHeight)),
+			attribute.Int("blob.count", len(blobs)),
+		),
+	)
+	defer span.End()
+
+	events := t.inner.ProcessBlobs(ctx, blobs, daHeight)
+	span.SetAttributes(attribute.Int("event.count", len(events)))
+	return events
 }

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@block/internal/syncing/da_retriever_tracing.go` around lines 67 - 69, The
tracing wrapper tracedDARetriever.ProcessBlobs is not creating any span or
attributes, so add instrumentation around the call to t.inner.ProcessBlobs:
start a span (using the same tracer used elsewhere in this file), set attributes
for blob_count (len(blobs)), da_height (daHeight) and optionally
total_blob_bytes (sum of len for each blob), call t.inner.ProcessBlobs(ctx,
blobs, daHeight), record the resulting events count as an attribute (e.g.,
da_events_count) and any error/status if applicable, then end the span; keep the
wrapper behavior identical except for adding the span and attributes to improve
visibility into latency and output volume.

apps/evm/server/force_inclusion_test.go (1)

53-58: Prefer a mockery-generated DA mock over extending the hand-written stub.

This local Subscribe stub works, but it increases interface-drift maintenance as DA APIs evolve. Using the generated DA mock keeps contract changes centralized.

As per coding guidelines, "Mock external dependencies using mockery".

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@apps/evm/server/force_inclusion_test.go` around lines 53 - 58, Replace the
hand-written Subscribe implementation on mockDA with the mockery-generated DA
mock: remove or stop using the local mockDA.Subscribe stub and instead import
and instantiate the mockery-created mock (e.g., MockDA) and use its EXPECT/On
setup to return a closed (<-chan da.SubscriptionEvent) or desired channel for
the test; update test setup where mockDA is constructed to use the mockery mock
and set the Subscribe return behavior via the mock's EXPECT/On methods so the
external DA contract is maintained centrally.

block/internal/da/tracing.go (1)

148-150: Consider tracing subscription setup for consistency.

Subscribe currently bypasses the tracing pattern used by the rest of this wrapper, so subscribe failures won’t be visible in DA spans.

Suggested refactor

 func (t *tracedClient) Subscribe(ctx context.Context, namespace []byte) (<-chan datypes.SubscriptionEvent, error) {
-	return t.inner.Subscribe(ctx, namespace)
+	ctx, span := t.tracer.Start(ctx, "DA.Subscribe",
+		trace.WithAttributes(
+			attribute.Int("ns.length", len(namespace)),
+			attribute.String("da.namespace", hex.EncodeToString(namespace)),
+		),
+	)
+	defer span.End()
+
+	ch, err := t.inner.Subscribe(ctx, namespace)
+	if err != nil {
+		span.RecordError(err)
+		span.SetStatus(codes.Error, err.Error())
+		return nil, err
+	}
+	return ch, nil
 }

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@block/internal/da/tracing.go` around lines 148 - 150, tracedClient.Subscribe
bypasses the wrapper's tracing; update the method to create a DA span (matching
the tracing pattern used elsewhere in this file), start the span with the
incoming ctx, add relevant attributes (e.g., namespace), call
t.inner.Subscribe(ctx, namespace), record any returned error on the span, and
end the span before returning the channel and error so subscription setup
failures are captured in DA spans; reference the tracedClient.Subscribe method
and t.inner.Subscribe call when making this change.

block/internal/syncing/da_follower.go (1)

265-279: ⚠️ Potential issue | 🔴 Critical

Use CAS-based rollback to avoid regressing localNextDAHeight.

Lines 265, 279, and 362 unconditionally Store(...) during rollback. If the other loop advanced localNextDAHeight meanwhile, this can move it backward and reprocess heights out of order.

🛠️ Proposed fix

 		for _, event := range events {
 			if err := f.pipeEvent(ctx, event); err != nil {
 				// Roll back so catchupLoop can retry this height.
-				f.localNextDAHeight.Store(ev.Height)
+				if !f.localNextDAHeight.CompareAndSwap(ev.Height+1, ev.Height) {
+					f.logger.Debug().
+						Uint64("da_height", ev.Height).
+						Uint64("local_next_da_height", f.localNextDAHeight.Load()).
+						Msg("skip rollback: localNextDAHeight already advanced")
+				}
 				f.logger.Warn().Err(err).Uint64("da_height", ev.Height).
 					Msg("failed to pipe inline event, catchup will retry")
 				return
 			}
 		}
@@
 		} else {
 			// No complete events (split namespace, waiting for other half).
-			f.localNextDAHeight.Store(ev.Height)
+			if !f.localNextDAHeight.CompareAndSwap(ev.Height+1, ev.Height) {
+				f.logger.Debug().
+					Uint64("da_height", ev.Height).
+					Uint64("local_next_da_height", f.localNextDAHeight.Load()).
+					Msg("skip rollback: localNextDAHeight already advanced")
+			}
 		}
@@
 		if err := f.fetchAndPipeHeight(ctx, local); err != nil {
 			// Roll back so we can retry after backoff.
-			f.localNextDAHeight.Store(local)
+			if !f.localNextDAHeight.CompareAndSwap(local+1, local) {
+				f.logger.Debug().
+					Uint64("da_height", local).
+					Uint64("local_next_da_height", f.localNextDAHeight.Load()).
+					Msg("skip rollback: localNextDAHeight already advanced")
+			}
 			if !f.waitOnCatchupError(ctx, err, local) {
 				return
 			}
 			continue
 		}

As per coding guidelines: "Be careful with concurrent access to shared state".

Also applies to: 360-363

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@block/internal/syncing/da_follower.go` around lines 265 - 279, The rollback
unconditionally calls f.localNextDAHeight.Store(...) which can regress the
counter if another goroutine advanced it; change these stores (the ones near
f.localNextDAHeight.Store in the rollback path and the other occurrences noted
around where events are handled and at the other block ~360-363) to a CAS
update: read cur := f.localNextDAHeight.Load(), and only attempt to set via a
CompareAndSwap (or a small CAS loop) to f.localNextDAHeight.CompareAndSwap(cur,
ev.Height) (or loop while cur < ev.Height and CAS fails: reload cur and retry)
so you only advance the value and never move it backward; apply this pattern to
each place that currently calls f.localNextDAHeight.Store(...) during rollback.

tools/local-da/local.go (1)

204-243: Consider extracting shared submit-to-rpc conversion logic.

SubmitWithOptions and Submit now duplicate the same namespace conversion, blob conversion, storage, and notify flow. A small shared helper would reduce drift risk between the two code paths.

Also applies to: 245-284

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@tools/local-da/local.go` around lines 204 - 243, SubmitWithOptions and Submit
duplicate namespace conversion, blob->rpc conversion, ID generation, storage
into d.data/d.blobData, timestamp/height bumping and d.notifySubscribers;
extract that shared logic into a single helper (e.g., processAndStoreBlobs or
storeBlobsAtNextHeight) that takes (ctx, ns []byte, blobs []datypes.Blob) and
performs: validate/convert namespace (libshare.NewNamespaceFromBytes), allocate
new height (increment d.height, set d.timestamps[d.height]=d.monotonicTime()),
generate IDs using d.nextID() and d.getHash(), append kvp to d.data[d.height],
build []*blobrpc.Blob via blobrpc.NewBlobV0 and set d.blobData[d.height], then
call d.notifySubscribers(d.height) and return the generated IDs (and error).
Replace the duplicated sections in SubmitWithOptions and Submit to call this
helper and keep logging/validation outside as appropriate.

block/internal/syncing/da_follower.go (3)

109-120: ⚠️ Potential issue | 🟠 Major

Catchup may never start if no subscription events arrive.

If the node starts behind and no new DA blob event arrives, catchupLoop will block indefinitely on catchupSignal. The signal is only sent by updateHighest, which is only called when subscription events arrive.

Consider seeding highestSeenDAHeight at startup by querying the latest DA height:

🐛 Proposed bootstrap fix

 func (f *daFollower) Start(ctx context.Context) error {
 	ctx, f.cancel = context.WithCancel(ctx)
 
+	// Seed highest DA height on startup so catchup can begin immediately
+	// even if no fresh subscription event arrives right away.
+	if f.client != nil {
+		if latest, err := f.client.GetLatestDAHeight(ctx); err == nil {
+			f.updateHighest(latest)
+		} else {
+			f.logger.Debug().Err(err).Msg("failed to seed latest DA height at startup")
+		}
+	}
+
 	f.wg.Add(2)
 	go f.followLoop(ctx)
 	go f.catchupLoop(ctx)

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@block/internal/syncing/da_follower.go` around lines 109 - 120, The
catchupLoop may never run if no subscription events arrive because
highestSeenDAHeight is only advanced via updateHighest; to fix, during
daFollower.Start seed highestSeenDAHeight (or call updateHighest) by querying
the latest DA height from the DA client before launching followLoop/catchupLoop
so catchupSignal can be triggered even with no incoming events; locate Start
(method daFollower.Start), highestSeenDAHeight, updateHighest, catchupSignal and
ensure Start obtains the current DA head and updates/sets highestSeenDAHeight
(or invokes updateHighest) and emits the catchupSignal if needed before starting
the goroutines.

---

260-281: ⚠️ Potential issue | 🟠 Major

Use CAS-based rollback to avoid regressing localNextDAHeight.

The unconditional Store(ev.Height) at Lines 265 and 279 can regress localNextDAHeight if catchupLoop has advanced it concurrently:

1. followLoop: CAS(N, N+1) succeeds
2. catchupLoop: CAS(N+1, N+2) succeeds
3. followLoop: pipeEvent fails → Store(N) regresses value from N+2 to N

This could cause duplicate processing of heights N and N+1.

🐛 Proposed CAS-based rollback

 		for _, event := range events {
 			if err := f.pipeEvent(ctx, event); err != nil {
-				// Roll back so catchupLoop can retry this height.
-				f.localNextDAHeight.Store(ev.Height)
+				// Roll back only if catchupLoop hasn't advanced past us.
+				f.localNextDAHeight.CompareAndSwap(ev.Height+1, ev.Height)
 				f.logger.Warn().Err(err).Uint64("da_height", ev.Height).
 					Msg("failed to pipe inline event, catchup will retry")
 				return
 			}
 		}
 		if len(events) != 0 {
 			if !f.headReached.Load() && f.localNextDAHeight.Load() > f.highestSeenDAHeight.Load() {
 				f.headReached.Store(true)
 			}
 			f.logger.Debug().Uint64("da_height", ev.Height).Int("events", len(events)).
 				Msg("processed subscription blobs inline (fast path)")
 		} else {
-			// No complete events (split namespace, waiting for other half).
-			f.localNextDAHeight.Store(ev.Height)
+			// No complete events (split namespace, waiting for other half).
+			// Roll back only if catchupLoop hasn't advanced past us.
+			f.localNextDAHeight.CompareAndSwap(ev.Height+1, ev.Height)
 		}

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@block/internal/syncing/da_follower.go` around lines 260 - 281, The rollback
after a failed pipeEvent (and the fallback when no complete events)
unconditionally calls f.localNextDAHeight.Store(ev.Height), which can regress
progress if catchupLoop advanced it; change both rollbacks to a CAS that only
sets ev.Height when the current value is ev.Height+1 (i.e., call
f.localNextDAHeight.CompareAndSwap(ev.Height+1, ev.Height)) so you only revert
if this goroutine still owns the +1 advance; do the same replacement for the
other Store(ev.Height) path, and optionally log or ignore the CAS failure
instead of forcing the store; locate these changes around the pipeEvent error
handling and the "No complete events" branch in the followLoop code that
manipulates localNextDAHeight.

---

362-369: ⚠️ Potential issue | 🟠 Major

Same CAS-based rollback issue in catchupLoop.

Line 364 has the same race condition as the inline processing path. If followLoop advances localNextDAHeight via inline processing between the CAS at Line 357 and the error rollback at Line 364, the Store will regress the value.

🐛 Proposed fix

 		if err := f.fetchAndPipeHeight(ctx, local); err != nil {
-			// Roll back so we can retry after backoff.
-			f.localNextDAHeight.Store(local)
+			// Roll back only if followLoop hasn't advanced past us.
+			f.localNextDAHeight.CompareAndSwap(local+1, local)
 			if !f.waitOnCatchupError(ctx, err, local) {
 				return
 			}

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@block/internal/syncing/da_follower.go` around lines 362 - 369, The rollback
in catchupLoop uses f.localNextDAHeight.Store(local) which can regress progress
if followLoop advanced the value; change the rollback to only set
localNextDAHeight back when the current value still equals the expected local
(use an atomic CompareAndSwap or conditional load+store): e.g., check
f.localNextDAHeight.Load() == local (or use
f.localNextDAHeight.CompareAndSwap(expectedLocal, local)) before performing the
Store, so fetchAndPipeHeight/followLoop races won't regress the counter.

test/testda/dummy.go (1)

298-311: Consider moving height increment inside the lock for consistency.

In Submit (line 149), d.height.Add(1) occurs inside the mutex-protected section. Here, it occurs before acquiring d.mu, creating a window where GetLatestDAHeight() returns the new height before the subscription event is sent. This ordering difference could cause subtle timing issues in tests.

♻️ Optional fix for consistency

 		select {
 		case <-ticker.C:
 			now := time.Now()
-			height := d.height.Add(1)
 			d.mu.Lock()
+			height := d.height.Add(1)
 			if d.headers[height] == nil {

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@test/testda/dummy.go` around lines 298 - 311, The height increment
d.height.Add(1) should be moved inside the mutex to match Submit's ordering and
avoid racing GetLatestDAHeight(): acquire d.mu before computing/assigning height
(inside the ticker handler), then update d.headers, call d.notifySubscribers
with the new height, and finally release the lock; ensure you still compute
Timestamp now := time.Now() as needed and keep notifySubscribers invoked while
the lock is held to preserve ordering consistency with Submit and avoid the
window where GetLatestDAHeight() could observe the new height before the
subscription event is emitted.

block/internal/syncing/da_follower.go (1)

88-106: Consider validating required config fields.

NewDAFollower doesn't validate that cfg.Client and cfg.Retriever are non-nil. If either is nil, subsequent calls in runSubscription or fetchAndPipeHeight will panic.

🛡️ Proposed defensive validation

 func NewDAFollower(cfg DAFollowerConfig) DAFollower {
+	if cfg.Client == nil {
+		panic("DAFollowerConfig.Client is required")
+	}
+	if cfg.Retriever == nil {
+		panic("DAFollowerConfig.Retriever is required")
+	}
+	if cfg.PipeEvent == nil {
+		panic("DAFollowerConfig.PipeEvent is required")
+	}
 	dataNs := cfg.DataNamespace
 	if len(dataNs) == 0 {
 		dataNs = cfg.Namespace
 	}

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@block/internal/syncing/da_follower.go` around lines 88 - 106, NewDAFollower
should validate required config fields: check that cfg.Client and cfg.Retriever
are non-nil before constructing the daFollower, returning an error or panicking
early as your codebase convention dictates; update NewDAFollower (and its caller
contract if needed) to validate DAFollowerConfig.Client and
DAFollowerConfig.Retriever and avoid creating a daFollower that will panic later
in runSubscription or fetchAndPipeHeight, ensuring any error message clearly
names the missing field(s) and references the constructor NewDAFollower and the
daFollower struct.

block/internal/syncing/da_follower.go (1)

261-282: ⚠️ Potential issue | 🟡 Minor

Unconditional Store can regress localNextDAHeight if catchupLoop advances concurrently.

Between the successful CAS at line 261 and the rollback at line 266 or 280, catchupLoop could claim and advance localNextDAHeight further. The unconditional Store(ev.Height) would then move the height backward, causing catchupLoop to re-process already-processed heights.

While the downstream processHeightEvent has duplicate detection, this causes unnecessary DA fetches and log noise.

🛡️ Proposed fix using CAS for rollback

 		for _, event := range events {
 			if err := f.pipeEvent(ctx, event); err != nil {
 				// Roll back so catchupLoop can retry this height.
-				f.localNextDAHeight.Store(ev.Height)
+				// Only roll back if we still own this height (no one advanced past us).
+				f.localNextDAHeight.CompareAndSwap(ev.Height+1, ev.Height)
 				f.logger.Warn().Err(err).Uint64("da_height", ev.Height).
 					Msg("failed to pipe inline event, catchup will retry")
 				return
 			}
 		}
 		if len(events) != 0 {
 			if !f.headReached.Load() && f.localNextDAHeight.Load() > f.highestSeenDAHeight.Load() {
 				f.headReached.Store(true)
 			}
 			f.logger.Debug().Uint64("da_height", ev.Height).Int("events", len(events)).
 				Msg("processed subscription blobs inline (fast path)")
 		} else {
 			// No complete events (split namespace, waiting for other half).
-			f.localNextDAHeight.Store(ev.Height)
+			// Only roll back if we still own this height.
+			f.localNextDAHeight.CompareAndSwap(ev.Height+1, ev.Height)
 		}

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@block/internal/syncing/da_follower.go` around lines 261 - 282, The rollback
currently does an unconditional f.localNextDAHeight.Store(ev.Height) which can
regress progress if catchupLoop advanced the height concurrently; change both
rollback sites (after pipeEvent error and the "No complete events" branch) to
perform a conditional compare-and-swap that only resets the height when the
current value is still ev.Height+1 (e.g., use
f.localNextDAHeight.CompareAndSwap(ev.Height+1, ev.Height)); this preserves a
rollback when we truly still own that slot but avoids moving the pointer
backward if catchupLoop has already advanced it. Ensure you update both
occurrences that reference f.localNextDAHeight.Store(ev.Height) and keep the
existing CAS used earlier as the pattern.